package io.renren.config;

import io.renren.common.utils.RedisUtils;
import org.springframework.amqp.AmqpException;
import org.springframework.amqp.core.*;
import org.springframework.amqp.rabbit.connection.ConnectionFactory;
import org.springframework.amqp.rabbit.connection.CorrelationData;
import org.springframework.amqp.rabbit.core.RabbitTemplate;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Qualifier;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;


@Configuration
public class RabbitConfig {

    public static final String QUEUE_GIFT = "ruyizhou.libao.queue";
    public static final String QUEUE_REBASE = "ruyizhou_rebase_queue";
    public static final String QUEUE_UPGRADE = "ruyizhou_upgrade_queue";
    public static final String QUEUE_FREEZE = "ruyizhou_freeze_queue";

    public static final String EXCHANGE_DIRECT_GIFT="ruyizhou.libao.exchange";
    public static final String EXCHANGE_DIRECT_REBASE="ruyizhou_rebase_exchange";
    public static final String EXCHANGE_DIRECT_UPGRADE="ruyizhou_upgrade_exchange";
    public static final String EXCHANGE_DIRECT_FREEZE="ruyizhou_freeze_exchange";

    public static final String ROUTINGKEY_GIFT="libao.verify.rebate";
    public static final String ROUTINGKEY_REBASE="ruyizhou_rebase_routingkey";
    public static final String ROUTINGKEY_UPGRADE="ruyizhou_upgrade_routingkey";
    public static final String ROUTINGKEY_FREEZE="ruyizhou_freeze_routingkey";

    @Autowired
    private RedisUtils redisUtils;

    @Bean(EXCHANGE_DIRECT_FREEZE)
    public Exchange EXCHANGE_DIRECT_FREEZE(){
        //durable(true) 持久化，mq重启之后交换机还在
        return ExchangeBuilder.directExchange(EXCHANGE_DIRECT_FREEZE).durable(true).build();
    }

    @Bean(QUEUE_FREEZE)
    public Queue QUEUE_FREEZE(){
        return new Queue(QUEUE_FREEZE);
    }

    //
    @Bean
    public Binding BINDING_ROUTINGKEY_FREEZE(@Qualifier(QUEUE_FREEZE) Queue queue,
                                           @Qualifier(EXCHANGE_DIRECT_FREEZE) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_FREEZE).noargs();
    }

    @Bean(EXCHANGE_DIRECT_GIFT)
    public Exchange EXCHANGE_DIRECT_GIFT(){
        //durable(true) 持久化，mq重启之后交换机还在
        return ExchangeBuilder.directExchange(EXCHANGE_DIRECT_GIFT).durable(true).build();
    }

    @Bean(QUEUE_GIFT)
    public Queue QUEUE_GIFT(){
        return new Queue(QUEUE_GIFT);
    }

    //
    @Bean
    public Binding BINDING_ROUTINGKEY_GIFT(@Qualifier(QUEUE_REBASE) Queue queue,
                                           @Qualifier(EXCHANGE_DIRECT_REBASE) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_REBASE).noargs();
    }

    @Bean(EXCHANGE_DIRECT_REBASE)
    public Exchange EXCHANGE_DIRECT_REBASE(){
        //durable(true) 持久化，mq重启之后交换机还在
        return ExchangeBuilder.directExchange(EXCHANGE_DIRECT_REBASE).durable(true).build();
    }

    @Bean(QUEUE_REBASE)
    public Queue QUEUE_REBASE(){
        return new Queue(QUEUE_REBASE);
    }

    //
    @Bean
    public Binding BINDING_ROUTINGKEY_REBASE(@Qualifier(QUEUE_REBASE) Queue queue,
                                             @Qualifier(EXCHANGE_DIRECT_REBASE) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_REBASE).noargs();
    }

    @Bean(EXCHANGE_DIRECT_UPGRADE)
    public Exchange EXCHANGE_DIRECT_UPGRADE(){
        //durable(true) 持久化，mq重启之后交换机还在
        return ExchangeBuilder.directExchange(EXCHANGE_DIRECT_UPGRADE).durable(true).build();
    }

    @Bean(QUEUE_UPGRADE)
    public Queue QUEUE_UPGRADE(){
        return new Queue(QUEUE_UPGRADE);
    }

    //
    @Bean
    public Binding BINDING_ROUTINGKEY_UPGRADE(@Qualifier(QUEUE_UPGRADE) Queue queue,
                                              @Qualifier(EXCHANGE_DIRECT_UPGRADE) Exchange exchange){
        return BindingBuilder.bind(queue).to(exchange).with(ROUTINGKEY_UPGRADE).noargs();
    } //

    @Bean
    @Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
    public RabbitTemplate rabbitTemplate(ConnectionFactory connectionFactory) {
        RabbitTemplate rabbitTemplate = new RabbitTemplate(connectionFactory);
        //消息发送到交换器时回调
        rabbitTemplate.setConfirmCallback(new RabbitTemplate.ConfirmCallback() {
            @Override
            public void confirm(CorrelationData correlationData, boolean b, String s) {
                System.out.println("redis key: " + correlationData.getId());
//                System.out.println(s);
                if(b) {
                    System.out.println("消息确认");
//                    redisUtils.delete(correlationData.getId());
//                    redisUtils.incr(correlationData.getId(), 1);
                    System.out.println(redisUtils.get(correlationData.getId()));
                } else {
                    //可以设置定时任务
                    System.out.println("try to resend msg");
                }
            }
        });
//        rabbitTemplate.setConfirmCallback((correlationData, ack, cause) -> {
//            if(ack) {
//                System.out.println("delete pre-msg in redis");
//            } else {
//                System.out.println(correlationData.toString());
//                System.out.println(s.toString());
//            }
//        });

        //消息发送到交换器，但无队列与交换器绑定时回调；
        rabbitTemplate.setReturnCallback(new RabbitTemplate.ReturnCallback() {
            @Override
            public void returnedMessage(Message message, int replyCode, String replyText, String exchange, String routingKey) {
                System.out.println(message.toString());
                System.out.println(replyCode);
                System.out.println(replyText);
                System.out.println(exchange);
                System.out.println(routingKey);
            }
        });
//        rabbitTemplate.setReturnCallback((message, replyCode, replyText, exchange, routingKey) -> {
//            System.out.println(message.toString());
//            System.out.println(replyCode);
//            System.out.println(replyText);
//            System.out.println(exchange);
//            System.out.println(routingKey);
//        });
        return rabbitTemplate;
    }

    @Bean
    public MessagePostProcessor correlationIdProcessor() {
        MessagePostProcessor messagePostProcessor = new MessagePostProcessor() {
            @Override
            public Message postProcessMessage(Message message, Correlation correlation) {
                MessageProperties messageProperties = message.getMessageProperties();

                if (correlation instanceof CorrelationData) {
                    String correlationId = ((CorrelationData) correlation).getId();
                    messageProperties.setCorrelationId(correlationId);
                }
                // 可以设置持久化，但与本文无关，因此没有附上
                return message;
            }

            @Override
            public Message postProcessMessage(Message message) throws AmqpException {
                return message;
            }
        };
        return messagePostProcessor;
    }
}